package unimelb.filesync.common.misc;

import java.util.concurrent.PriorityBlockingQueue;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.ReentrantLock;

public class BoundedPriorityBlockingQueue<E> extends PriorityBlockingQueue<E> {

    private static final long serialVersionUID = -2517781037540862034L;

    private final int capacity;

    private final ReentrantLock lock;

    private final Condition notFull;

    private final Condition notEmpty;

    public BoundedPriorityBlockingQueue(int capacity) {
        super(capacity);
        this.capacity = capacity;
        this.lock = new ReentrantLock(true);
        this.notFull = lock.newCondition();
        this.notEmpty = lock.newCondition();
    }

    @Override
    public void put(E e) {
        try {
            try {
                lock.lockInterruptibly();
                while (this.size() == capacity) {
                    notFull.await();
                }
            } catch (InterruptedException exception) {
                notFull.signal();
                throw new RuntimeException(exception);
            }
            super.put(e);
            notEmpty.signal();
        } finally {
            lock.unlock();
        }
    }

    @Override
    public E take() throws InterruptedException {
        try {
            lock.lockInterruptibly();
            try {
                while (size() == 0)
                    notEmpty.await();
            } catch (InterruptedException ie) {
                notEmpty.signal(); // propagate to non-interrupted thread
                throw ie;
            }
            E element = super.take();
            notFull.signal();
            return element;
        } finally {
            lock.unlock();
        }
    }

}
